package com.lkinga.www.Util;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.lkinga.www.Do.JobInitModel;

import java.io.IOException;

/**
 * Created by xiaohei on 16/2/22.
 * 椹卞姩绋嬪簭璁剧疆绫�,鍦ㄥ悇涓猰r鐨勯┍鍔ㄧ▼搴忎腑璋冪敤姝ょ被鐨処nitJob鏂规硶鍗冲彲鍒濆鍖朖ob
 */
public class BaseDriver {
    /**
     * 鏍规嵁涓嶅悓mr浣滀笟鐨勪笉鍚屽弬鏁板垵濮嬪寲Job鎻愪緵浣跨敤
     *
     * @param jobs 涓�涓垨澶氫釜淇濆瓨鐫�job淇℃伅鐨凧obInitModel瀵硅薄
     *             TODO:鍙牴鎹渶瑕佽繘琛屼慨鏀�,濡傛坊鍔犺嚜瀹氫箟鐨勫垎缁�,鎺掑簭,瑙勭害绫荤殑璁剧疆
     */
    public static void initJob(JobInitModel[] jobs
    ) throws IOException, ClassNotFoundException, InterruptedException {
        for (JobInitModel jobInitModel : jobs) {
            //妫�鏌dfs涓婄殑杈撳嚭鐩綍
            FileSystem fs = FileSystem.get(jobInitModel.getConf());
            Path fileOutPath = new Path(jobInitModel.getOutPath());
            if (fs.exists(fileOutPath)) {
                fs.delete(fileOutPath, true);
            }

            //鍒濆鍖杍ob
            Job job;
            if (jobInitModel.getJob() == null) {
                job = Job.getInstance(jobInitModel.getConf(), jobInitModel.getJobName());
            } else {
                job = jobInitModel.getJob();
            }
            job.setJarByClass(jobInitModel.getJarClass());
            //璁剧疆杈撳叆璺緞鍜宖ormat绫�
            String[] inPathsStr = jobInitModel.getInPaths();
            Path[] inPaths = new Path[inPathsStr.length];
            for (int i = 0; i < inPathsStr.length; i++) {
                inPaths[i] = new Path(inPathsStr[i]);
            }
            FileInputFormat.setInputPaths(job, inPaths);
            if (jobInitModel.getInputFormatClass() != null) {
                job.setInputFormatClass(jobInitModel.getInputFormatClass());
            } else {
                job.setInputFormatClass(TextInputFormat.class);
            }

            //mapper绫荤浉鍏宠缃�
            job.setMapperClass(jobInitModel.getMapper());
            job.setMapOutputKeyClass(jobInitModel.getMapOutKeyClass());
            job.setMapOutputValueClass(jobInitModel.getMapOutValueClass());

            //Partitioner璁剧疆
            if (jobInitModel.getPartitionerClass() != null) {
                job.setPartitionerClass(jobInitModel.getPartitionerClass());
            }

            //combiner璁剧疆
            if (jobInitModel.getCombinerClass() != null) {
                job.setCombinerClass(jobInitModel.getCombinerClass());
            }

            if (jobInitModel.getReducer() != null) {
                //reducer绫荤浉鍏宠缃�
                job.setReducerClass(jobInitModel.getReducer());
                job.setOutputKeyClass(jobInitModel.getReduceOutKeyClass());
                job.setOutputValueClass(jobInitModel.getReduceOutValueClass());
            }

            //璁剧疆杈撳嚭鐩綍鍜岃緭鍑虹殑format绫�
            FileOutputFormat.setOutputPath(job, fileOutPath);
            job.setOutputFormatClass(TextOutputFormat.class);

            //鎻愪氦浣滀笟
            job.waitForCompletion(true);
        }
    }
}
